package com.ruoyi.kafka;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

/**
 * @program: ruoyi
 * @description: StudyTestPartition
 * @author: zengchen
 * @create: 2020-10-19 09:48
 **/
public class StudyTestPartition implements Partitioner {

    /**
     * Compute the partition for the given record.
     *
     * @param topic      The topic name
     * @param key        The key to partition on (or null if no key)
     * @param keyBytes   The serialized key to partition on( or null if no key)
     * @param value      The value to partition on or null
     * @param valueBytes The serialized value to partition on or null
     * @param cluster    The current cluster metadata
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        String keyStr = (String) key;
        String lastNum = keyStr.substring(keyStr.lastIndexOf("-") + 1);
        if (Integer.parseInt(lastNum) > 9) {
            return 2;
        }
        return Integer.parseInt(lastNum) % 2;
    }

    /**
     * This is called when partitioner is closed.
     */
    @Override
    public void close() {

    }

    /**
     * Configure this class with the given key-value pairs
     *
     * @param configs
     */
    @Override
    public void configure(Map<String, ?> configs) {

    }
}
